<?php
namespace YCore;

use Workerman\Lib\Timer;
use Workerman\Worker;

/**
 * 队列处理进程类
 * @author YeWJ <yeweijian299@163.com>
 */
class QueueWorker extends Worker
{
    /**
     * 版本号
     */
    const Q_VERSION  = '1.0.0';
    /**
     * 任务列表
     * @var array
     */
    public $queues   = [];
    /**
     * 轮询间隔时间
     * @var integer
     */
    public $interval = 1;
    /**
     * 当前执行的任务
     * @var Job
     */
    protected $currentJob;

    public function run()
    {
        $this->onWorkerStart = [$this, 'onWorkerStart'];
        parent::run();
    }

    public function onWorkerStart()
    {
        /** 利用定时器轮询队列任务 */
        Timer::add($this->interval, [$this, 'doWork']);
    }

    public function doWork()
    {
        while ($job = $this->reserve()) {
            /** 任务开始 */
            $this->workingOn($job);
            /** 处理任务 */
            $this->perform($job);
            /** 任务结束 */
            $this->doneWorking();
        }
        return true;
    }

    public function perform(Job $job)
    {
        try {
            $job->perform();
        } catch (\Exception $e) {
            $this->qLog($job . ' failed: ' . $e->getMessage());
            $job->fail($e);
            return;
        }

        $job->updateStatus(Job\Status::STATUS_COMPLETE);
        $this->qLog('done ' . $job);
    }

    public function reserve()
    {
        $queues = $this->queues();
        if (!is_array($queues)) {
            return false;
        }
        foreach ($queues as $queue) {
            // $this->qLog('Checking ' . $queue);
            $job = Job::reserve($queue);
            if ($job) {
                $this->qLog('Found job on ' . $queue);
                return $job;
            }
        }

        return false;
    }

    public function queues($fetch = true)
    {
        if (!in_array('*', $this->queues) || $fetch == false) {
            return $this->queues;
        }

        $queues = YQueue::queues();
        sort($queues);
        return $queues;
    }

    public function workingOn(Job $job)
    {
        $job->worker      = $this;
        $this->currentJob = $job;
        $job->updateStatus(Job\Status::STATUS_RUNNING);
        $data = json_encode([
            'queue'   => $job->queue,
            'run_at'  => strftime('%a %b %d %H:%M:%S %Z %Y'),
            'payload' => $job->payload,
        ]);
        YQueue::redis()->set('worker:' . $job->worker, $data);
    }

    public function doneWorking()
    {
        $this->currentJob = null;
        YQueue::redis()->del('worker:' . (string) $this);
    }

    public function qLog($msg)
    {
        if (self::$daemonize) {
            echo $msg . PHP_EOL;
            return;
        }

        $date = date('Y-m-d H:i:s');
        $path = __DIR__;

        if (defined('LOGSPATH')) {
            $path = LOGSPATH;
        }
        $path = rtrim($path, '/') . "/queue_log/{$this->name}-{$this->id}/";

        if (!is_dir($path)) {
            mkdir($path, 0755, true);
        }

        error_log("[{$date}] " . $msg . PHP_EOL, 3, $path . strstr($date, ' ', true) . ".log");
    }

    public function __toString()
    {
        return (string) $this->id;
    }
}
